package com.lagou.hbase_hw;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class HbaseClient {

    static Configuration conf = null;
    static Connection conn = null;

    public static void init() throws IOException {

        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","linux121,linux122");
        conf.set("hbase.zookeeper.property.clientPort","2181");
        conn = ConnectionFactory.createConnection(conf);
    }

    // 创建Hbase表
    public void createTable() throws IOException {

        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();

        HTableDescriptor friend = new HTableDescriptor(TableName.valueOf("friend"));

        friend.addFamily(new HColumnDescriptor("friends"));

        admin.createTable(friend);

        System.out.println("friend表创建成功");

        admin.close();

    }

    public void putData() throws IOException {

        Table friend = conn.getTable(TableName.valueOf("friend"));

        Put put1 = new Put(Bytes.toBytes("uid1"));
        put1.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid2"),Bytes.toBytes("uid2"));
        put1.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid3"),Bytes.toBytes("uid3"));

        Put put2 = new Put(Bytes.toBytes("uid2"));
        put2.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid1"),Bytes.toBytes("uid1"));
        put2.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid3"),Bytes.toBytes("uid3"));

        Put put3 = new Put(Bytes.toBytes("uid3"));
        put3.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid1"),Bytes.toBytes("uid1"));
        put3.addColumn(Bytes.toBytes("friends"),Bytes.toBytes("uid2"),Bytes.toBytes("uid2"));

        friend.put(put1);
        friend.put(put2);
        friend.put(put3);

        friend.close();

    }

    public void deleteFriend(String u1, String u2) throws IOException {

        Table friend = conn.getTable(TableName.valueOf("friend"));
        Get get = new Get(Bytes.toBytes(u1));
        get.addFamily(Bytes.toBytes("friends"));
        Result res = friend.get(get);
        Cell[] cells = res.rawCells();
        for (Cell cell : cells) {
            if(Bytes.toString(CellUtil.cloneValue(cell)).equals(u2)) {
                Delete delete = new Delete(Bytes.toBytes(u1));
                delete.addColumn(Bytes.toBytes("friends"), Bytes.toBytes(u2));
                friend.delete(delete);
                deleteFriend(u2, u1);
            }
        }
        return;
    }

    public void realse(){

        if(conn!=null){
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws IOException {

        HbaseClient.init();
        HbaseClient hc = new HbaseClient();
        hc.createTable();
        hc.putData();
        hc.deleteFriend("uid1", "uid2");
        hc.realse();
    }
}
